#include "thread_pool.h"

/*init a thread_pool*/
static Thread_pool *pool = NULL;
int pool_init(unsigned int thread_num)
{
    unsigned int i;

    pool = (Thread_pool *)malloc(thread_num*sizeof(Thread_pool) );
    if( NULL == pool )
    {
        return -1;
    }
    pthread_mutex_init(&(pool->pool_lock),NULL);
    pthread_cond_init(&(pool->job_ready), NULL );
    pool->jobs = NULL;
    pool->thread_num=thread_num;
    pool->size=0;
    pool->destory=0;
    pool->threads=(pthread_t *)malloc(thread_num * sizeof(pthread_t) );
    
    for( i=0;i<thread_num;i++)
    {
      pthread_create(&(pool->threads[i]),NULL,routine,NULL);
    }
    return 0;
}

/*add a job into thread pool*/
int pool_add_job(void *(*process)(void *),void *arg)
{
	Job *tmp;
    Job * newjob=(Job *)malloc(sizeof(Job));
    assert(process != NULL );
    newjob->process=process;
    assert(newjob->process != NULL );
    newjob->arg=arg;
    newjob->next=NULL;
    pthread_mutex_lock(&(pool->pool_lock));
    tmp = pool->jobs;
    if(tmp != NULL )
    {
        while(tmp->next != NULL )
        {
            tmp=tmp->next;
        }
        tmp->next=newjob;
    } 
    else
    {
        pool->jobs=newjob;
    }
    assert(pool->jobs != NULL );
    assert(pool->jobs->process != NULL );
    pool->size++;
    pthread_mutex_unlock(&(pool->pool_lock));
    pthread_cond_signal(&(pool->job_ready));
    return 0;
}

/*destroy the thread pool*/
int pool_destroy(void)
{
    unsigned int i;
    Job *head=NULL;
    /*already destory*/
    if( pool->destory)
    {
        return -1;
    }
    /*notify all threads*/
    pthread_cond_broadcast( &(pool->job_ready));
    for(i=0;i<pool->thread_num;i++)
    {
        pthread_join(pool->threads[i],NULL);
    }
    free(pool->threads);
    while(pool->jobs)
    {
        head=pool->jobs;
        pool->jobs=pool->jobs->next;
        free(head);
    }
    pthread_mutex_destroy(&(pool->pool_lock));
    pthread_cond_destroy(&(pool->job_ready));
    free(pool);
    pool=NULL;
    return 0;
}

/*every thread call this function*/
void *routine(void *arg)
{
    Job *job;
    mlog(POOLLOGFILE, _FL_, "start thread %lu\n",pthread_self());
    while(1)
    {
        pthread_mutex_lock(&(pool->pool_lock));
        while(pool->size == 0 && !pool->destory)
        {
            mlog("demo.log", _FL_, "thread %lu is waiting\n", pthread_self());
            pthread_cond_wait( &(pool->job_ready),&(pool->pool_lock) );
        }
        if(pool->destory)
        {
            pthread_mutex_unlock(&(pool->pool_lock));
            mlog("demo.log", _FL_,"thread %lu will exit\n",pthread_self());
            pthread_exit(NULL);
        }
        mlog(POOLLOGFILE, _FL_, "thread %lu is starting to work\n", pthread_self() );
        assert(pool->size != 0);
        assert(pool->jobs != NULL );
        pool->size--;
        job=pool->jobs;
        pool->jobs=job->next;
        pthread_mutex_unlock(&(pool->pool_lock));
        /*process job*/
        (*(job->process))(job->arg);
        free(job);
        job=NULL;
    }
    /*never get here*/
    pthread_exit(NULL);
}

